package com.webchat.pgc.messagequeue.consumer;

import com.webchat.common.enums.messagequeue.MessageBroadChannelEnum;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.enums.search.DataSourceTypeEnum;
import com.webchat.common.service.messagequeue.consumer.AbstractRedisDelayQueueConsumer;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.JsonUtil;
import com.webchat.domain.dto.queue.ArticleDelayMessageDTO;
import com.webchat.domain.dto.search.SyncSearchEngineDTO;
import com.webchat.domain.dto.search.SyncSearchEngineListDTO;
import com.webchat.domain.vo.response.publicaccount.ArticleBaseResponseVO;
import com.webchat.pgc.repository.dao.IArticleDAO;
import com.webchat.pgc.service.OfficialArticleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 公众号推送MQ消费
 */
@Service
public class ArticlePushDelayQueueConsumer extends AbstractRedisDelayQueueConsumer<ArticleDelayMessageDTO> {

    @Autowired
    private MessageQueueProducer<Set<String>, Long> messageQueueProducer;

    @Autowired
    private MessageQueueProducer<SyncSearchEngineListDTO, Long> syncSearchEngineMessageQueueProducer;

    @Autowired
    private IArticleDAO articleDAO;

    @Autowired
    private OfficialArticleService officialArticleService;

    @Override
    protected ArticleDelayMessageDTO convert(String s) {

        return JsonUtil.fromJson(s, ArticleDelayMessageDTO.class);
    }

    @Override
    protected MessageQueueEnum getMessageQueue() {

        return MessageQueueEnum.QUEUE_OFFICIAL_ARTICLE_PUSH_MESSAGE;
    }

    @Override
    protected void doNextConsume(Set<String> messages) {

        Set<ArticleDelayMessageDTO> articleDelayMessages = messages.stream().map(this::convert).collect(Collectors.toSet());

        // 走redis批量查询文章详情
        List<Long> articleIdList = articleDelayMessages.stream().map(ArticleDelayMessageDTO::getArticleId).toList();
        Map<Long, ArticleBaseResponseVO> articleDetailMap = officialArticleService.batchGetArticleDetailFromCache(articleIdList);
        /**
         * 构造同步搜索引擎业务数据
         */
        List<SyncSearchEngineDTO> syncDataList = articleDelayMessages.stream().map(am -> {
            ArticleBaseResponseVO article = articleDetailMap.get(am.getArticleId());
            SyncSearchEngineDTO syncSearchEngineDTO = new SyncSearchEngineDTO();
            syncSearchEngineDTO.setPk(String.valueOf(am.getArticleId()));
            syncSearchEngineDTO.setSummary(String.format("《%s》: %s", article.getTitle(), article.getDescription()));
            syncSearchEngineDTO.setContent(article.getContent());
            syncSearchEngineDTO.setSourceType(DataSourceTypeEnum.ARTICLE.getSource());
            return syncSearchEngineDTO;
        }).collect(Collectors.toList());
        SyncSearchEngineListDTO syncData = new SyncSearchEngineListDTO();
        syncData.setDataList(syncDataList);
        syncSearchEngineMessageQueueProducer.send(MessageQueueEnum.QUEUE_SYNC_SEARCH_DB_MESSAGE, syncData);

        /**
         * 1. 广播通知connect service完成推文、以及消息持久化，批量提交任务到普通队列
         */
        messageQueueProducer.broadSend(MessageBroadChannelEnum.QUEUE_ARTICLE_PUSH_MESSAGE, messages);
        /**
         * 2. 更新文章已经推送状态
         */
        Set<Long> articleIds = articleDelayMessages.stream()
                .map(ArticleDelayMessageDTO::getArticleId).collect(Collectors.toSet());
        articleDAO.updatePushedStatus(articleIds);
        /**
         * 3. 持久化消息: 由connect服务来完成，在推送成功之后
         */
    }
}
